package com.yuyuan.notice.event;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSONObject;

import com.yuyuan.notice.domain.Event;
import com.yuyuan.notice.utils.CommunityConstant;
import com.yuyuan.notice.mapper.NoticeMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import com.yuyuan.notice.domain.po.Notice;

@Component
public class EventNoticeConsumer implements CommunityConstant {

    private static final Logger logger = LoggerFactory.getLogger(EventNoticeConsumer.class);


    @Autowired
    private NoticeMapper noticeMapper;

    @KafkaListener(topics = {TOPIC_COMMENT, TOPIC_LIKE, TOPIC_FOLLOW})
    public void handleSendMessage(ConsumerRecord record) {
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空！");
            return;
        }
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {
            logger.error("消息格式错误！");
            return;
        }
        noticeMapper.insert(BeanUtil.copyProperties(event.getData(), Notice.class));
    }
}
